
#include <string.h>
#include <unistd.h>

#include "../../common/log.h"
#include "../coder/tinypb_coder.h"
#include "../fd_event_group.h"
#include "../rpc/rpc_dispatcher.h"
#include "tcp_connection.h"

namespace rocket {

TcpConnection::TcpConnection(
    EventLoop *event_loop, int fd, int buffer_size, NetAddr::s_ptr local_addr,
    NetAddr::s_ptr peer_addr,
    TcpConnectionType type /* = TcpConnectionByServer*/)
    : m_event_loop(event_loop), m_local_addr(local_addr),
      m_peer_addr(peer_addr), m_state(NotConnected), m_fd(fd),
      m_connection_type(type) {
    m_in_buffer = std::make_shared<TcpBuffer>(buffer_size);
    m_out_buffer = std::make_shared<TcpBuffer>(buffer_size);
    DEBUGLOG("m_out_buffer size %d", m_out_buffer->getBufferSize());
    m_fd_event = FdEventGroup::GetFdEventGroup()->getFdEvent(fd);
    m_fd_event->setNonBlock();

    m_coder = new TinyPBCoder();

    if (m_connection_type == TcpConnectionByServer) {
        listenRead();
    }

    // m_coder = new
}

TcpConnection::~TcpConnection() {
    DEBUGLOG("~TcpConnection");
    if (m_coder) {
        delete m_coder;
        m_coder = nullptr;
    }
}

void TcpConnection::setState(const TcpConnection::TcpState state) {
    m_state = state;
}

TcpConnection::TcpState TcpConnection::getState() { return m_state; }

void TcpConnection::onRead() {

    if (m_state != Connected) {
        ERRORLOG("onRead error, read from NotConnected client, addr[%s], "
                 "clientfd[%d]",
                 m_peer_addr->toString(), m_fd);
        return;
        // exit(1);
    }

    bool is_read_all = false;
    bool is_closed = false;
    while (!is_read_all) {
        if (m_in_buffer->writable() == 0) {
            m_in_buffer->resizeBuffer(2 * m_in_buffer->getBufferSize());
        }
        int read_count = m_in_buffer->writable();
        // int write_index = m_in_buffer->writeIndex();

        char tmp[read_count];
        memset(&tmp[0], 0, sizeof(char) * read_count);
        int rt = ::read(m_fd, &tmp[0], read_count);

        if (rt > 0) {
            DEBUGLOG("success read %d bytes from addr[%s], client fd[%d]", rt,
                     m_peer_addr->toString().c_str(), m_fd);
            m_in_buffer->writeToBuffer(&tmp[0], rt);
            if (rt == read_count) {
                continue;
            } else if (rt < read_count) {
                is_read_all = true;
                break;
            }
        } else if (rt == 0) {
            is_closed = true;
            break;
        } else if (rt == -1 && errno == EAGAIN) {
            is_read_all = true;
            break;
        }
    }

    if (is_closed) {
        INFOLOG("peer closed, peer addr [%s], clientfd [%d]",
                m_peer_addr->toString().c_str(), m_fd);
        clear();
        return;
    }

    if (!is_read_all) {
        ERRORLOG("not read all data, peer addr [%s], clientfd [%d]",
                 m_peer_addr->toString().c_str(), m_fd);
    }
    // TODO: 协议解析
    execute();
}

void TcpConnection::execute() {

    if (m_connection_type == TcpConnectionByServer) {

        // 完成 RPC 请求执行业务逻辑，获得响应并返回
        std::vector<AbstractProtocol::s_ptr> result;
        std::vector<AbstractProtocol::s_ptr> reply_messages;

        m_coder->decode(result, m_in_buffer);

        for (size_t i = 0; i < result.size(); ++i) {
            // 1. Get response message from every request (invoke rpc function)
            // 2. put message into buffer, listen EPOLLOUT package;
            INFOLOG("success get request[%s] from client[%s]",
                    result[i]->m_req_id.c_str(),
                    m_peer_addr->toString().c_str());

            std::shared_ptr<TinyPBProtocol> message =
                std::make_shared<TinyPBProtocol>();

            // message->m_pb_data = "hello, this is rocket rpc test data";
            // message->m_req_id = result[i]->m_req_id;
            RpcDispatcher::GetRpcDispatcher()->dispatch(result[i], message,
                                                        this);

            reply_messages.emplace_back(message);
        }

        m_coder->encode(reply_messages, m_out_buffer);

        listenWrite();
    } else {
        // 从 buffer decode 得到 message 对象，执行回调函数
        std::vector<AbstractProtocol::s_ptr> result;
        m_coder->decode(result, m_in_buffer);

        for (size_t i = 0; i < result.size(); i++) {
            std::string req_id = result[i]->m_req_id;
            auto it = m_read_dones.find(req_id);
            if (it != m_read_dones.end()) {
                it->second(result[i]);
            }
        }
    }
}

void TcpConnection::onWrite() {

    if (m_state != Connected) {
        ERRORLOG("onWrite error, read from NotConnected client, addr[%s], "
                 "clientfd[%d]",
                 m_peer_addr->toString().c_str(), m_fd);
    }

    if (m_connection_type == TcpConnectionByClient) {
        // write message to buffer
        // 1. encode message
        std::vector<AbstractProtocol::s_ptr> messages;
        for (size_t i = 0; i < m_write_dones.size(); i++) {
            messages.push_back(m_write_dones[i].first);
        }

        m_coder->encode(messages, m_out_buffer);

        // 2. write message
    }

    bool is_write_all = false;
    while (true) {
        if (m_out_buffer->readable() == 0) {
            DEBUGLOG("no data is needed to be sent to [%s]",
                     m_peer_addr->toString().c_str());
            is_write_all = true;
            break;
        }
        int write_size = m_out_buffer->readable();
        // int read_index = m_out_buffer->readIndex();

        std::vector<char> tmp;
        tmp.resize(write_size);
        DEBUGLOG("test 1");
        m_out_buffer->readFromBuffer(tmp, write_size);

        int rt = ::write(m_fd, &tmp[0], write_size);

        if (rt >= write_size) {
            DEBUGLOG("no data is needed to be sent to [%s]",
                     m_peer_addr->toString().c_str());
            is_write_all = true;
            break;
        }
        if (rt == -1 && errno == EAGAIN) {
            // send buffer is full
            // wait until the next time to write.
            ERRORLOG("write data error, errno==EAGIAN, rt == -1");
            break;
        }
    }
    if (is_write_all) {
        m_fd_event->cancel(FdEvent::OUT_EVENT);
        m_event_loop->addEpollEvent(m_fd_event);
    }

    if (m_connection_type == TcpConnectionByClient) {
        for (size_t i = 0; i < m_write_dones.size(); i++) {
            m_write_dones[i].second(m_write_dones[i].first);
        }
    }
    m_write_dones.clear();
}

void TcpConnection::clear() {
    // 进行连接关闭后的清理行为

    if (m_state == Closed) {
        return;
    }
    m_fd_event->cancel(FdEvent::IN_EVENT);
    m_fd_event->cancel(FdEvent::OUT_EVENT);

    m_event_loop->deleteEpollEvent(m_fd_event);

    m_state = Closed;
}

void TcpConnection::shutDown() {
    if (m_state == Closed || m_state == NotConnected) {
        return;
    }

    m_state = HalfClosing;
    ::shutdown(m_fd, SHUT_RDWR);
}

void TcpConnection::setConnectionType(TcpConnection::TcpConnectionType type) {
    m_connection_type = type;
}

void TcpConnection::listenWrite() {
    m_fd_event->listen(FdEvent::OUT_EVENT,
                       std::bind(&TcpConnection::onWrite, this));

    m_event_loop->addEpollEvent(m_fd_event);
}

void TcpConnection::listenRead() {
    m_fd_event->listen(FdEvent::IN_EVENT,
                       std::bind(&TcpConnection::onRead, this));

    m_event_loop->addEpollEvent(m_fd_event);
}

void TcpConnection::pushSendMessage(
    AbstractProtocol::s_ptr message,
    std::function<void(AbstractProtocol::s_ptr)> done) {
    m_write_dones.push_back(std::make_pair(message, done));
}

void TcpConnection::pushReadMessage(
    const std::string &req_id,
    std::function<void(AbstractProtocol::s_ptr)> done) {
    m_read_dones.insert(std::make_pair(req_id, done));
}

NetAddr::s_ptr TcpConnection::getLocalAddr() { return m_local_addr; }

NetAddr::s_ptr TcpConnection::getPeerAddr() { return m_peer_addr; }
} // namespace rocket
